package im.vinci.server.common.tmsg;

import com.aliyun.openservices.ons.api.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.DefaultTransactionStatus;

import javax.sql.DataSource;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;

/**
 * Created by tim@vinci on 17/2/9.
 */
public class SingleDBTransactionManager extends DataSourceTransactionManager{

    private static final Logger log = LoggerFactory.getLogger(SingleDBTransactionManager.class);

    private final DataSource defDataSource;

    private final SingleDBMessageStore store;

    private final Producer producer;

    public SingleDBTransactionManager(DataSource dataSource, Producer producer)
    {
        super.setDataSource(dataSource);
        this.defDataSource = dataSource;
        this.store = new SingleDBMessageStore(dataSource);
        this.producer = producer;
    }

    @Override
    public void afterPropertiesSet() {
        super.afterPropertiesSet();
        new Timer("TransactionMsgScanTimerTask").scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                boolean b = false;
                try {
                    b = store.tryLock();
                    if (!b) {
                        return;
                    }
                    List<ProducerMessage> list = store.selectTimeoutMsg();
                    if (list == null) {
                        return;
                    }
                    for (ProducerMessage msg : list) {
                        sendMsg(msg);
                    }
                }finally {
                    if (b) {
                        store.deleteLock();
                    }
                }

            }
        },5000,10000);
    }

    @Override
    protected void doBegin(Object transaction, TransactionDefinition definition)
    {
        TransactionMessageHolder.init(this.store, this);
        super.doBegin(transaction, definition);
    }

    @Override
    protected void doCommit(DefaultTransactionStatus status)
    {
        super.doCommit(status);
        List<ProducerMessage> list = TransactionMessageHolder.clear();
        if (list != null) {
            for (ProducerMessage msg : list) {
                sendMsg(msg);
            }
        }
    }

    private void sendMsg(ProducerMessage msg) {
        if (msg == null) {
            return;
        }
        try {
            producer.send(msg.toMessage());
            store.finish(msg);
            log.info("消息发送成功{}",msg.getKey());
        } catch (Throwable t) {
            log.error("消息发送失败{}", msg.getKey(), t);
            try {
                store.error(msg);
            }catch (Exception e) {
                log.error("标记发送失败消息失败{}", msg.getKey(),e);
                //ignore
            }
        }
    }

    @Override
    protected void doCleanupAfterCompletion(Object transaction)
    {
        super.doCleanupAfterCompletion(transaction);
        List<ProducerMessage> list = TransactionMessageHolder.clear();
        if (list != null)
            for (ProducerMessage msg : list)
                log.info("事务提交失败, 消息({})被忽略.subject:{}", msg.getKey(), msg.getTopic());
    }

}
